package com.youth.config.kafka;

import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.*;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;

import java.util.Map;

/**
 * @ClassName KafkaConfig
 * @Description: TODO
 * @Author yjwu   https://www.whatyun.cn/article/434
 * @Date 2020/1/16 16:29
 **/
@Configuration
public class KafkaConfig {

    @Bean
    @Primary
    @ConfigurationProperties(prefix = "spring.kafka")
    public KafkaProperties kafkaProperties() {
        return new KafkaProperties();
    }

    /**
     * @MethodName: kafkaListenerContainerFactory
     * @Description: TODO 监听容器工厂
     * @Param: * @param
     * @Return: {@link KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String,String>>}
     * @Author: yjwu
     * @Date: 2020/1/16 16:42
     **/
    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        // 设置消费者工厂
        factory.setConsumerFactory(consumerFactory());
        // 要创建的消费者数量(10 个线程并发处理)
        factory.setConcurrency(2);
        factory.getContainerProperties().setPollTimeout(3000);
        return factory;
    }

    /**
     * @MethodName: consumerFactory
     * @Description: TODO 消费者工厂
     * @Param: * @param
     * @Return: {@link ConsumerFactory< String, String>}
     * @Author: yjwu
     * @Date: 2020/1/16 16:42
     **/
    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<String, String>(consumerConfigs());
    }

    private Map<String, Object> consumerConfigs() {
        return kafkaProperties().buildConsumerProperties();
    }

    /**
     * @return
     * @Description: KafkaTemplate
     * @Author:杨攀
     * @Since: 2019年7月10日下午2:10:47
     */
    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<String, String>(producerFactory());
    }

    /**
     * @MethodName: producerFactory
     * @Description: TODO 生产者工厂
     * @Param: * @param
     * @Return: {@link ProducerFactory< String, String>}
     * @Author: yjwu
     * @Date: 2020/1/16 16:42
     **/
    @Bean
    public ProducerFactory<String, String> producerFactory() {
        return new DefaultKafkaProducerFactory<String, String>(producerConfigs());
    }

    private Map<String, Object> producerConfigs() {
        return kafkaProperties().buildProducerProperties();
    }


}
